package cn.edu.flink.scala.tutorial.table

import cn.edu.flink.scala.tutorial.source.ClickSource
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Expressions.$
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._

object SelectTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val clickStream = env.addSource(new ClickSource)

    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    val clickTable: Table = tableEnv.fromDataStream(clickStream)

    val resultTable: Table = clickTable.select($("url"), $("user")).where($("user").isEqual("Mary"))

    tableEnv.toDataStream(resultTable).print("table")

    env.execute("TableTest")
  }
}
